/**
 *
 */
package com.gomeplus.commons.quartz.netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.*;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.gomeplus.commons.quartz.build.TaskBuildSource;
import com.gomeplus.commons.quartz.cache.QuartzConstants;
import com.gomeplus.commons.quartz.cache.QuartzPropertiesCache;
import com.gomeplus.commons.quartz.cache.ServerIpCache;
import com.gomeplus.commons.quartz.factory.TaskListenerClientFactory;
import com.gomeplus.commons.quartz.netty.NettyMessage;

/**
 * @author chenmin-ds
 * netty客户端的启动类
 */
public class NettyClientStartup{

	private Logger logger = LoggerFactory.getLogger("NettyClientStartup");

	private EventLoopGroup workGroup = null;
	private Bootstrap bootstrap = null;
	private ChannelFuture channelFuture = null;
	private TaskListenerClientFactory taskListenerClientFactory;
	private TaskBuildSource taskBuildSourceProxy;
	private static class Singleton {
		private static final NettyClientStartup singleton = new NettyClientStartup();
	}

	public static final NettyClientStartup getInstance() {
		return Singleton.singleton;
	}

	public void setBusinessObject(TaskListenerClientFactory taskListenerClientFactory){
		this.taskListenerClientFactory = taskListenerClientFactory;
	}

	public void setTaskBuildSource(TaskBuildSource taskBuildSourceProxy){
		this.taskBuildSourceProxy = taskBuildSourceProxy;
	}


	public synchronized void stop(){
		if(channelFuture != null){
			channelFuture.channel().close();
		}
		if(workGroup != null){
			workGroup.shutdownGracefully();
		}
	}

	public synchronized void start(){
		bootstrap = new Bootstrap();
		workGroup = new NioEventLoopGroup();
		bootstrap.group(workGroup).channel(NioSocketChannel.class);
		bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
		bootstrap.option(ChannelOption.TCP_NODELAY, true);
		bootstrap.handler(new ChannelInitializer<Channel>() {

			@Override
			protected void initChannel(Channel channel) throws Exception {
				ChannelPipeline channelPipeline = channel.pipeline();
				channelPipeline.addLast(new ObjectDecoder(1024*1024, ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
				channelPipeline.addLast(new ObjectEncoder());
				channelPipeline.addLast(new IdleStateHandler(QuartzPropertiesCache.getReaderIdleTimeSeconds(),
						QuartzPropertiesCache.getWriterIdleTimeSeconds(),QuartzPropertiesCache.getAllIdleTimeSeconds(),TimeUnit.SECONDS));
				channelPipeline.addLast("nettyClientHeartBeatHandler",new NettyClientHeartBeatHandler(taskBuildSourceProxy));
				channelPipeline.addLast("nettyClientTaskInfoHandler" , new NettyClientTaskInfoHandler(taskListenerClientFactory));
			}

		});
		doConnect();
		Runtime.getRuntime().addShutdownHook(new Thread(){
			@Override
			public void run() {
				workGroup.shutdownGracefully();
			}
		});
	}

	public Channel getChannel(){
		if(channelFuture != null){
			return channelFuture.channel();
		}else {
			return  null;
		}
	}


	public void doConnect(){
		if(null == channelFuture || channelFuture.isDone()){
			if(null != channelFuture && channelFuture.isDone()){
				channelFuture.channel().close();
			}
			ChannelFutureListener futureListener = new ChannelFutureListener(){
				@Override
				public void operationComplete(ChannelFuture f) {
					if (f.isSuccess()) {
						channelFuture= f;
						String localAddress = f.channel().localAddress().toString().replace("/", "");
						logger.info("客户端启动后的地址为：localAddress=" + localAddress);
						String[] localAddressArray = localAddress.split(":");
						ServerIpCache.setNettyClientIp(localAddressArray[0]);
						ServerIpCache.setNettyClientPort(Integer.valueOf(localAddressArray[1]));
						clientRegister( f.channel());
						//重联后马上把信息推送到服务端 2017-09-22 陈民
						new NettyClientTaskRunLogSend().sendInfoServer();
					} else {
						f.channel().eventLoop().schedule(new Runnable() {
							@Override
							public void run() {
								logger.info("重新连接服务端");
								doConnect();
							}
						}, 3, TimeUnit.SECONDS);
					}
				}
			};
			channelFuture = bootstrap.connect(ServerIpCache.getNettyServerIp(), ServerIpCache.getNettyServerPort()).addListener(futureListener);
		}
	}

	private void clientRegister(Channel channel){
		try{
			NettyMessage nettyMessage = new NettyMessage(QuartzConstants.CLIENT_REGISTER);
			nettyMessage.setJsonObject("BUSINESS_CODE", QuartzPropertiesCache.getBusinessCode());
			channel.writeAndFlush(nettyMessage);
		}catch(Exception e){
			logger.error("netty创建客户端与服务端建立连接后，向服务端发起注册请求异常，IP【{}】port【{}】",ServerIpCache.getNettyServerIp(),ServerIpCache.getNettyServerPort());
		}
	}


	static  class  Task implements Runnable{
		private CountDownLatch countDownLatch;
		private CyclicBarrier cyclicBarrier;
		Task(CountDownLatch cdl ,CyclicBarrier cb){
			this.countDownLatch = cdl;
			this.cyclicBarrier = cb;
		}
		@Override
		public void run() {
			try {
				// System.out.println("start netty client" );
				cyclicBarrier.await();
				NettyClientStartup nettyClient = new NettyClientStartup();
				// String ip = "10.144.32.174";
				//  String ip = "10.144.35.117";
				String ip = "10.112.179.41";
				//String ip = "127.0.0.1";
				int port =10440;
				ServerIpCache.setNettyServerIp(ip);
				ServerIpCache.setNettyServerPort(port);
				nettyClient.start();
				countDownLatch.countDown();
				// System.out.println("------"+ addr);
				NettyMessage nettyMessage = new NettyMessage(QuartzConstants.BUSINESS_TYPE_XT);
				nettyMessage.setJsonObject("BUSINESS_CODE", QuartzPropertiesCache.getBusinessCode());
				for (int i=1;i<1000;i++){
					nettyClient.getChannel().writeAndFlush(nettyMessage);
				}
			}catch (Exception e){
				System.out.println("eerror" + e);
			}
		}
	}

	public static void main(String[] args) throws Exception {
		NettyClientStartup nettyClient = new NettyClientStartup();
		String ip = "10.133.125.110";
		int port =10441;
		ServerIpCache.setNettyServerIp(ip);
		ServerIpCache.setNettyServerPort(port);
		nettyClient.start();
		nettyClient.channelFuture.channel().close();
		System.out.println("1111");
	}
}
